package worker

import (
	"crontab/common"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"time"
)

type JobMgr struct {
	client *zk.Conn
	lease  time.Duration
	path   string
}

//全局单例
var (
	G_JobMgr *JobMgr
)

func callback(event zk.Event) {

	fmt.Println("*******************")
	fmt.Println("path:", event.Path)
	fmt.Println("type:", event.Type.String())
	fmt.Println("state:", event.State.String())
	fmt.Println("-------------------")
}

func InitJobMgr() (err error) {
	//option := zk.WithEventCallback(callback)
	con, connChan, err := zk.Connect(G_Config.ZKEndPoint, time.Duration(G_Config.ZKConnectTimeout)*time.Millisecond)

	if err != nil {

		return err
	}

	for {
		isConnect := false
		select {
		case connEvent := <-connChan:
			if connEvent.State == zk.StateConnected {
				isConnect = true
				fmt.Println("connect to zookeeper server success")
			}
		case _ = <-time.After(time.Duration(G_Config.ZKConnectTimeout) * time.Millisecond):
			return errors.New("connect to zookeeper server timeout")

		}
		if isConnect {

			break
		}
	}
	G_JobMgr = &JobMgr{
		client: con,
		lease:  time.Duration(G_Config.LockTimeout) * time.Millisecond,
		path:   G_Config.LockPath,
	}
	//创建根节点,"/cron/jobs"
	if _, err := G_JobMgr.CreatePath(common.JOB, []byte(""), 0); err == nil {
		if _, err := G_JobMgr.CreatePath(common.JOB_SVAE_PATH, []byte(""), 0); err == nil {
			if _, err := G_JobMgr.CreatePath(common.JOB_KILLER_PATH, []byte(""), 0); err == nil {
				if _, err := G_JobMgr.CreatePath(common.JOB_LOCK_PATH, []byte(""), 0); err == nil {
					return nil
				}
			}
		}
	}

	G_JobMgr.client.Close()

	return err
}

func InitWatcher() (err error) {
	err = G_JobMgr.WatchKillerJobsW()
	err = G_JobMgr.WatchRunningJobsW()
	return
}

func (jobMgr *JobMgr) IsExsist(path string) (b bool, err error) {
	b, _, err = jobMgr.client.Exists(path)
	return b, err
}

func (jobMgr *JobMgr) CreatePath(path string, data []byte, flag int32) (str string, err error) {

	exists, err := jobMgr.IsExsist(path)
	if err != nil {
		return "", err
	}
	if !exists {
		str, err = jobMgr.client.Create(path, data, flag, zk.WorldACL(zk.PermAll))
		if err != nil && err != zk.ErrNodeExists {
			return "", err
		}

	}
	return path, err
}

func (jobMgr *JobMgr) SetData(path string, data []byte) (err error) {
	if _, err = jobMgr.CreatePath(path, data, 0); err == zk.ErrNodeExists || err == nil {

		jobMgr.client.Set(path, data, -1)
		return nil
	}

	return err
}

func (jobMgr *JobMgr) DeleteData(path string) (err error) {
	err = jobMgr.client.Delete(path, -1)

	return err
}

func (jobMgr *JobMgr) GetData(path string) (value []byte, err error) {
	if value, _, err = jobMgr.client.Get(path); err != nil {
		value = nil
	}

	return value, err
}

func (jobMgr *JobMgr) GetChildrens(path string) (value []string, err error) {

	if value, _, err = jobMgr.client.Children(path); err != nil {
		value = nil
	}

	return value, err
}

func (jobMgr *JobMgr) SaveJob(job *common.Job) (oldJob *common.Job, err error) {
	//把任务保存到指定路径下
	var (
		jobPath  string
		jobValue []byte
		oldValue []byte
	)
	jobPath = common.JOB_SVAE_PATH + "/" + job.Name
	//任务信息
	if jobValue, err = json.Marshal(job); err != nil {
		return
	}
	oldValue, err = jobMgr.GetData(jobPath)

	if err = jobMgr.SetData(jobPath, jobValue); err != nil {
		return
	}

	if oldValue != nil {
		if err = json.Unmarshal(oldValue, &oldJob); err != nil {
			err = nil
			return
		}
	}

	//如果更新成功
	return oldJob, err
}

func (jobMgr *JobMgr) DeleteJob(jobPath string) (oldJob *common.Job, err error) {
	//把任务保存到指定路径下
	var (
		oldValue []byte
	)

	//先要进行查询任务信息
	oldValue, err = jobMgr.GetData(jobPath)
	if err != nil {
		return nil, err
	}

	if err = jobMgr.DeleteData(jobPath); err != nil {
		return nil, err
	}
	if oldValue != nil {
		if err = json.Unmarshal(oldValue, &oldJob); err != nil {
			err = nil
			return
		}
	}

	//如果更新成功
	return oldJob, err
}

func (jobMgr *JobMgr) ListJob(path string) (oldValue []string, cnt int, err error) {
	//把任务保存到指定路径下

	if path == "" {
		path = common.JOB_SVAE_PATH
	}

	//先要进行查询任务信息
	if oldValue, err = jobMgr.GetChildrens(path); err != nil {
		return nil, 0, err
	}

	//如果更新成功
	return oldValue, cnt, err
}

//原理就是想/cron/killer中添加一个任务
func (jobMgr *JobMgr) KillJob(name string) (err error) {

	if _, err = jobMgr.DeleteJob(common.JOB_SVAE_PATH + "/" + name); err != nil {
		return err
	}

	err = jobMgr.SetData(common.JOB_KILLER_PATH+"/"+name, []byte(name))
	return err
}

func (jobMgr *JobMgr) ZkStateStringFormat(s *zk.Stat) string {
	return fmt.Sprintf("Czxid:%d\nMzxid: %d\nCtime: %d\nMtime: %d\nVersion: %d\nCversion: %d\nAversion: %d\nEphemeralOwner: %d\nDataLength: %d\nNumChildren: %d\nPzxid: %d\n",
		s.Czxid, s.Mzxid, s.Ctime, s.Mtime, s.Version, s.Cversion, s.Aversion, s.EphemeralOwner, s.DataLength, s.NumChildren, s.Pzxid)
}

func notify(chann <-chan zk.Event) {
	fmt.Println("Before *******************")
	event := <-chann
	fmt.Println("After*******************")
	fmt.Println("path:", event.Path)
	fmt.Println("type:", event.Type.String())
	fmt.Println("state:", event.State.String())
	fmt.Println("-------------------")
}

//worker监听节点的变化
func (jobMgr *JobMgr) WatchRunningJobsW() (err error) {
	var (
		jobs []string

		jobEvent *common.JobEvent
		JobMaps  map[string]*common.Job
		jobItem  []byte
	)
	JobMaps = make(map[string]*common.Job)
	jobs, _, _, err = jobMgr.client.ChildrenW(common.JOB_SVAE_PATH)

	for _, value := range jobs {
		if jobItem, _, _, err = jobMgr.client.GetW(common.JOB_SVAE_PATH + "/" + value); err == nil {
			oldJob := &common.Job{}
			if err = json.Unmarshal(jobItem, &oldJob); err != nil {
				err = nil
				continue
			} else {
				jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, oldJob)
				JobMaps[oldJob.Name] = oldJob

				//把这个任务推送给调度协程
				G_Scheduler.SendJobEventToScheduler(jobEvent)
			}
		} else {

		}

	}

	go func() {
		var (
			watcher <-chan zk.Event
			ret     *common.Job
			isExist bool
		)
		for {

			if jobItem, _, watcher, err = jobMgr.client.GetW(common.JOB_WATCHER_PATH + "/" + "curd"); err == nil {
				select {
				case event := <-watcher:
					if event.Type == zk.EventNodeDataChanged {
						if ret, err = common.UnPackJson(jobItem); err != nil {

						} else {
							if isExist, _, _ = jobMgr.client.Exists(common.JOB_SVAE_PATH + "/" + ret.Name); isExist {
								jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, ret)
							} else {
								jobEvent = common.BuildJobEvent(common.JOB_EVENT_DELETE, ret)
							}

							G_Scheduler.SendJobEventToScheduler(jobEvent)
						}

					}
				}
			} else {
				time.Sleep(time.Second * 1)
			}

		}
	}()

	return err
}

//worker监听killer节点的变化--我们只关注killer下面的创建事件
func (jobMgr *JobMgr) WatchKillerJobsW() (err error) {

	go func() {
		var (
			watcher  <-chan zk.Event
			ret      *common.Job
			isExist  bool
			jobItem  []byte
			jobEvent *common.JobEvent
		)
		for {

			if jobItem, _, watcher, err = jobMgr.client.GetW(common.JOB_WATCHER_PATH + "/" + "killer"); err == nil {
				select {
				case event := <-watcher:
					if event.Type == zk.EventNodeDataChanged {
						if ret, err = common.UnPackJson(jobItem); err != nil {

						} else {
							if isExist, _, _ = jobMgr.client.Exists(common.JOB_KILLER_PATH + "/" + ret.Name); isExist {
								jobEvent = common.BuildJobEvent(common.JOB_EVENT_KILLER, ret)
								G_Scheduler.SendJobEventToScheduler(jobEvent)
							}

						}

					}
				}
			} else {
				time.Sleep(time.Second * 1)
			}
		}

	}()
	return err
}

//创建分布式执行锁
func (jobMgr *JobMgr) CreateDistributeLock(jobName string) (jobLock *common.JobRedisDistributeLock) {
	jobLock = common.InitDistributeLock(jobName, jobMgr.path, jobMgr.lease)
	return
}

//创建分布式执行锁
func (jobMgr *JobMgr) CreateZKDistributeLock(jobName string) (jobLock *common.JobZKDistributeLock) {
	jobLock = common.InitZKDistributeLock(G_JobMgr.client, jobName, jobMgr.path, jobMgr.lease)
	return
}
